
#include "app_pool.h"
#include "stdio.h"
#include "mqueue.h"
#include "pthread.h"
#include "log.h"
// 创建消息队列的描述符
static mqd_t mqd;
static pthread_t threads[APP_POOL_THREAD_NUM];
/**
 * 线程执行的函数
 */
void *app_pool_thread_func(void *arg)
{
    task_t task;
    while (1)
    {
        // 线程主动拉区任务
        int size = mq_receive(mqd, (char *)&task, sizeof(task), 0);
        if (size > 0)
        {

            task.task_func(task.task_arg);
        } /* code */
    }
}

/**
 * 初始化线程池
 */

int app_pool_init()
{
    mq_unlink(APP_POOL_QUEUE_NAME);//如果已经有队列，新的队列不会被创建，所以要删除

    struct mq_attr attr;
    attr.mq_maxmsg = APP_POOL_QUEUE_MSGSIZE;
    attr.mq_msgsize = sizeof(task_t);

    // 1.创建队列                      创建和读写权限
    mqd = mq_open(APP_POOL_QUEUE_NAME, O_CREAT | O_RDWR, 0666, &attr);
    if (mqd == -1)
    {
        perror("mq_open");
        return -1;
    }
    log_info("创建线程池队列完成");
    // 1.创建线程池
    for (size_t i = 0; i < APP_POOL_THREAD_NUM; i++)
    {
        pthread_create(&threads[i], NULL, app_pool_thread_func, NULL);
        log_info("创建一个线程成功");
    }
    return 0;
}

/**
 * 向线程池提交任务   task :1.函数 2.参数
 */
int app_pool_submit_task(void *(*func)(void *), void *arg)
{
    // 1.封装任务
    task_t task =
        {
            .task_arg = arg,
            .task_func = func};
    // 2.向线程池提交任务
    int ret = mq_send(mqd, (char *)&task, sizeof(task), 0);
    if (ret == -1)
    {
        perror("mq_send");
        return -1;
    }
    else
    {
        log_info("提交任务成功");
    }
    return 0;
}

/**
 * 销毁线程池
 */
int app_pool_destroy(){
    //1. 线程池结束掉
    for(size_t i = 0; i < APP_POOL_THREAD_NUM; i++){
        //取消线程
        pthread_cancel(threads[i]);
    }
    for (size_t i = 0; i < APP_POOL_THREAD_NUM; i++)
    {   //等待线程结束
        pthread_join(threads[i], NULL);
    }
    log_info("线程池销毁成功 , 已结束所有线程");
    //2.队列关闭
    mq_close(mqd);
    // 删除队列
    mq_unlink(APP_POOL_QUEUE_NAME);
    log_info("队列关闭成功 , 删除队列");
    return 0;
}